package com.coding4play.mqtt.store.impl;

import cn.hutool.core.util.StrUtil;
import com.coding4play.mqtt.common.entity.message.SubMessage;
import com.coding4play.mqtt.common.repository.ISubMessageRepository;
import com.coding4play.mqtt.store.config.PropertiesConfig;
import com.coding4play.mqtt.store.config.RedisTemplateWrapper;
import com.coding4play.mqtt.store.util.RedisScan;
import com.coding4play.mqtt.store.util.StringTopicMatch;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 订阅存储服务实现
 */
@Component
public class SubMessageRepositoryImpl extends RedisTemplateWrapper implements ISubMessageRepository {

    /**
     * hash结构，[topic, (clientId, msg)] 存放订阅了此topic的客户端
     * 包含通配符的topic
     */
    private final static String HASH_SUB_WILDCARD_PRE = "mqtt:sub_wildcard:topic_clientId_msg:";
    /**
     * hash结构，[topic, (clientId, msg)] 存放订阅了此topic的客户端
     * 不含通配符的topic
     */
    private final static String HASH_SUB_NOT_WILDCARD_PRE = "mqtt:sub_notWildcard:topic_clientId_msg:";
    /**
     * set结构，存放此client订阅的topic
     */
    private final static String SET_SUB_CLIENT_TOPIC_PRE = "mqtt:sub:clientId_topic:";

    private PropertiesConfig propertiesConfig;


    public SubMessageRepositoryImpl(RedisTemplate<String, Object> redisTemplate,
                                    PropertiesConfig propertiesConfig) {
        super(redisTemplate);
        this.propertiesConfig = propertiesConfig;
    }

    @Override
    public void put(String topicFilter, SubMessage subMessage) {
        String key;
        String clientId = subMessage.getClientId();

        if (StrUtil.contains(topicFilter, '#') || StrUtil.contains(topicFilter, '+')) {
            key = getKey(HASH_SUB_WILDCARD_PRE, topicFilter);
        } else {
            key = getKey(HASH_SUB_NOT_WILDCARD_PRE, topicFilter);
        }
        HASH.put(key, clientId, subMessage);
        SET.add(getKey(SET_SUB_CLIENT_TOPIC_PRE, clientId), topicFilter);
    }

    @Override
    public void delete(String topicFilter, String clientId) {
        String key;
        if (StrUtil.contains(topicFilter, '#') || StrUtil.contains(topicFilter, '+')) {
            key = getKey(HASH_SUB_WILDCARD_PRE, topicFilter);
        } else {
            key = getKey(HASH_SUB_NOT_WILDCARD_PRE, topicFilter);
        }
        HASH.delete(key, clientId);
        redisTemplate.delete(getKey(SET_SUB_CLIENT_TOPIC_PRE, clientId));
    }

    @Override
    public void delete(String clientId) {
        String clientKey = getKey(SET_SUB_CLIENT_TOPIC_PRE, clientId);
        Set<Object> members = SET.members(clientKey);

        members.forEach(topic -> {
            HASH.delete(getKey(HASH_SUB_WILDCARD_PRE, (String) topic), clientId);
            HASH.delete(getKey(HASH_SUB_NOT_WILDCARD_PRE, (String) topic), clientId);
        });
        redisTemplate.delete(clientKey);
    }

    /**
     * 返回订阅了参数topic的客户端订阅集合
     *
     * @param topic
     * @return
     */
    @Override
    public List<SubMessage> search(String topic) {

        //首先找一下无通配的
        List<SubMessage> msgList = new ArrayList<>();
        Collection<Object> objects = HASH.entries(getKey(HASH_SUB_NOT_WILDCARD_PRE, topic)).values();
        List<SubMessage> nwList = objects.stream()
                .map(obj -> (SubMessage) obj)
                .collect(Collectors.toList());
        msgList.addAll(nwList);

        /**
         * 再找通配的
         */
        // 1，先由前缀scan下所有的key
        List<String> keyList = RedisScan.keyScan(
                redisTemplate.getConnectionFactory(), propertiesConfig.isRedisCluster, getKey(HASH_SUB_WILDCARD_PRE, "*")
        );

        // 2，对key进行匹配，因为有【“+”， “#”】通配，过滤后再去redis查找
        int length = HASH_SUB_WILDCARD_PRE.length();
        List<SubMessage> wList = keyList.stream()
                .filter(k -> StringTopicMatch.match(topic, k.substring(length)))
                .flatMap(k -> HASH.values(k).stream())
                .map(obj -> (SubMessage) obj)
                .collect(Collectors.toList());
        msgList.addAll(wList);

        return msgList;
    }


    private String getKey(String pre, String append) {
        return pre + append;
    }

}
